package com.learn.window;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @create: 2023-04-19 16:06
 * @author: Mr.Du
 * --------------
 * @notes: 滚动计数窗口
 **/
public class TumblingCountWindowDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> streamSource = env.socketTextStream("node1", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> operator = streamSource.map(word -> Tuple2.of(word, 1)).returns(Types.TUPLE(Types.STRING, Types.INT));
        //count window all
        operator.countWindowAll(5)
                        .sum(1).print("窗口内的单词总数为：");

        //统计每个分组后的数据，每隔5条
        //这个5条是每个分组内达到5条，不是总计达到5条
        operator.keyBy(0)
                        .countWindow(5)
                                .sum(1).print("窗口内对应单词个数为：");
        env.execute();
    }
}
